package com.pai4j.pgc.consumer;

import com.pai4j.common.enums.messagequeue.MessageBroadChannelEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.service.messagequeue.consumer.AbstractRedisDelayQueueConsumer;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.domain.dto.queue.ArticleDelayMessageDTO;
import com.pai4j.pgc.dao.IArticleDAO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Set;
import java.util.stream.Collectors;

/**
 * 专题文章推送MQ消费
 *
 * @author: CYM-pai
 * @date: 2025/04/29 20:59
 **/
@Service
public class ArticlePushDelayQueueConsumer extends AbstractRedisDelayQueueConsumer<ArticleDelayMessageDTO> {

    @Autowired
    private MessageQueueProducer<Set<String>, Long> messageQueueProducer;

    @Autowired
    private IArticleDAO articleDAO;

    @Override
    protected ArticleDelayMessageDTO convert(String s) {
     //将json字符串转换为ArticleDelayMessageDTO对象
        return JsonUtil.fromJson(s, ArticleDelayMessageDTO.class);
    }
   /**
     * 获取消息队列
     *
     * @return
     */
    @Override
    protected MessageQueueEnum getMessageQueue() {

        return MessageQueueEnum.QUEUE_SUBJECT_ARTICLE_PUSH_MESSAGE;
    }

    
    /**
     * 实际的消费消息方法
     *
     * @param messages
     */
    @Override
    protected void doNextConsume(Set<String> messages) {

        /**
         * 1. 广播通知connect service完成推文、以及消息持久化，批量提交任务到普通队列
         * 这里从延迟队列中获取到的消息，需要广播到普通队列中，由connect service来完成，避免高并发下推文失败
         */
        messageQueueProducer.broadSend(MessageBroadChannelEnum.QUEUE_ARTICLE_PUSH_MESSAGE, messages);
        /**
         * 2. 更新文章已经推送状态
         */
        // 将消息列表转换为ArticleDelayMessageDTO的集合
        // 使用流处理来遍历每个消息，并调用convert方法进行转换，最后收集到一个Set集合中
        //this::convert是一个方法引用，表示调用"当前类"中的convert方法。
        Set<ArticleDelayMessageDTO> articleDelayMessages = messages.stream().map(this::convert).collect(Collectors.toSet());
        // 获取文章ID的集合
        Set<Long> articleIds = articleDelayMessages.stream()
                .map(ArticleDelayMessageDTO::getArticleId).collect(Collectors.toSet());
        articleDAO.updatePushedStatus(articleIds);
        /**
         * 3. 持久化消息: 由connect服务来完成，在推送成功之后
         */
        
        
    }
}
